use crate::parser::*;
use std::{
    collections::HashMap,
    io,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf},
    net::{self, TcpStream},
    sync::{
        oneshot::{self},
        Mutex,
    },
};

/// client 端实例，
/// client实现功能相对比较单一,就是能够向服务器pub消息,然后就会说订阅消息,
/// 订阅的主题收到消息以后能够得到通知.因此总结起来就是下面三个功能:
/// 提供pub接口
/// 提供sub接口
/// 处理sub后收到的消息

static _SID: AtomicU64 = AtomicU64::new(0);
type MessageHandler = Box<dyn FnMut(&[u8]) -> std::result::Result<(), ()> + Sync + Send>;

// #[derive(Debug)]
pub struct Client {
    addr: String,
    sid: u64,
    pub stop: Option<oneshot::Sender<()>>,
    writer: Arc<Mutex<WriteHalf<TcpStream>>>,
    handler: Arc<Mutex<HashMap<String, MessageHandler>>>,
}

impl Client {

    pub fn get_addr(&self) -> &str {
        &self.addr
    }

    /// 连接服务器
    pub async fn connect<I: Into<String>>(addr: I) -> io::Result<Self> {
        let addr = addr.into();
        let conn = net::TcpStream::connect(&addr).await?;
        let (r, w) = tokio::io::split(conn);
        let (tx, rx) = tokio::sync::oneshot::channel();
        let client = Client {
            addr: addr,
            sid: _SID.fetch_add(1, Ordering::Relaxed),
            stop : Some(tx),
            writer: Arc::new(Mutex::new(w)),
            handler: Arc::new(Default::default()),
        };
        let handler = client.handler.clone();
        let writer = client.writer.clone();
        tokio::spawn(async move { Self::receive_task(r, rx, handler, writer).await });
        Ok(client)
    }

    /// 发送消息 PUB subject size\r\n{message}
    pub async fn push_message(&mut self, subject: &str, msg: &[u8]) -> io::Result<()> {
        let mut writer = self.writer.lock().await;
        let m = format!("PUB {} {}\r\n", subject, msg.len());
        let _ = writer.write_all(m.as_bytes()).await;
        let _ = writer.write_all(msg).await;
        writer.write_all("\r\n".as_bytes()).await
    }

    /// 向服务器发布一条sub消息,然后等待服务器推送相关消息.
    //sub消息格式为SUB subject {queue} {sid}\r\n
    //可能由于rustc的bug,导致如果subject是&str,则会报错E0700,暂时使用String来替代
    pub async fn sub_message(
        &mut self,
        subject: String,
        queue: Option<String>,
        handler: MessageHandler,
    ) -> std::io::Result<()> {
        self.sid += 1;
        let mut writer = self.writer.lock().await;
        let m = if let Some(queue) = queue {
            format!("SUB {} {} {}\r\n", subject.as_str(), queue, self.sid)
        } else {
            format!("SUB {} {}\r\n", subject.as_str(), self.sid)
        };
        self.handler.lock().await.insert(subject, handler);
        writer.write_all(m.as_bytes()).await
    }

    /// receive_task主要是做消息的接收,解析,以及将消息派发给合适的handler.
    /// 这个其实是本模块最复杂的地方,总体上比较直观.
    /// 主要有以下两点
    /// 使用futures::select这个宏来辅助实现同时监控多个future
    /// TcpStream如果read到size为0,说明连接已经关闭,无需继续
    async fn receive_task(
        mut reader: ReadHalf<TcpStream>,
        stop: oneshot::Receiver<()>,
        handler: Arc<Mutex<HashMap<String, MessageHandler>>>,
        writer: Arc<Mutex<WriteHalf<TcpStream>>>,
    ) {
        let mut buf = [0u8; 512];
        let mut parser = Parser::new();
        use futures::*;
        let mut stop = stop.fuse();
        loop {
            select! {
                _=stop=>{
                    println!("client closed");
                    return;
                }
                r = reader.read(&mut buf[..]).fuse()=>{
                     let n = {
                        match r {
                            Err(e) => {
                                println!("read err {}", e);
                                let _ = writer.lock().await.shutdown().await;
                                return;
                            }
                            Ok(n) => n,
                        }
                    };
                    if n == 0 {
                        //EOF,说明对方关闭了连接
                        return;
                    }
                    let mut buf2 = &buf[..n];
                    loop {
                        let r = parser.parse(buf2);
                        let (r, n) = match r {
                            Err(e) => {
                                println!("parse err {}", e);
                                let _ = writer.lock().await.shutdown().await;
                                return;
                            }
                            Ok(r) => r,
                        };
                        //                println!("receive msg {:?}", r);
                        match r {
                            ParseResult::NoMsg => {
                                break;
                            }
                            ParseResult::MsgArg(msg) => {
                                Self::process_message(msg, &handler).await;
                                parser.clear_msg_buf();
                            }
                        }
                        //缓冲区处理完毕
                        if n == buf.len() {
                            break;
                        }
                        buf2 = &buf2[n..]
                    }
                }
            }
        }
    }

    /*
    根据消息的subject，找到订阅方，
    然后推送给他们
    */
    pub async fn process_message(msg: MsgArg<'_>, handler: &Arc<Mutex<HashMap<String, MessageHandler>>>) {
        //        println!("broadcast msg {}", msg.subject);
        let mut handler = handler.lock().await;
        let h = handler.get_mut(msg.subject);
        if let Some(h) = h {
            let _ = h(msg.msg);
        }
    }
}

#[cfg(test)]
mod tests {

    use super::*;

    #[test]
    fn test_new_client() {
        let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_io()
        .build()
        .unwrap();

        let c = Client::connect("127.0.0.1:4222");
        let c = rt.block_on(c).unwrap();
        println!("{:?}", c.get_addr());

        let c = Client::connect("127.0.0.1:4222".to_owned());
        let c = rt.block_on(c).unwrap();
        println!("{:?}", c.get_addr());
    }
}
